package com.iakuil.app.common.redis.base;

import com.iakuil.app.common.core.constant.RedisConstant;
import org.springframework.dao.QueryTimeoutException;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.Record;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;

import java.util.Objects;

/**
 * Redis StreamListener with key
 *
 * @author Kai
 */
public interface RedisStreamListener<K, V extends Record<K, ?>> extends StreamListener<K, V> {

    /**
     * 监听的Stream的key
     *
     * @return Stream的key
     */
    default String key() {
        return RedisConstant.DEFAULT_STREAM_KEY;
    }

    /**
     * 消费组
     *
     * @return 消费组
     */
    default String group() {
        return RedisConstant.DEFAULT_STREAM_GROUP;
    }

    /**
     * 消费者名称
     *
     * @return 消费者名称
     */
    default String consumer() {
        return RedisConstant.DEFAULT_STREAM_CONSUMER;
    }

    /**
     * 订阅Stream的key和获取偏移量，以及消费处理类
     *
     * @return StreamReadRequest
     */
    default StreamMessageListenerContainer.StreamReadRequest<String> getRequest() {
        StreamOffset<String> streamOffset = StreamOffset.create(Objects.toString(key(), RedisConstant.DEFAULT_STREAM_KEY), ReadOffset.lastConsumed());
        // 每个Subscription配置一个消费者
        Consumer consumer = Consumer.from(Objects.toString(group(), RedisConstant.DEFAULT_STREAM_GROUP), Objects.toString(consumer(), RedisConstant.DEFAULT_STREAM_CONSUMER));
        return StreamMessageListenerContainer.StreamReadRequest.builder(streamOffset).cancelOnError(throwable -> {
            // 重要：防止查询超时中断监听！
            return !(throwable instanceof QueryTimeoutException);
        }).consumer(consumer).autoAcknowledge(false).build();
    }
}
